package service

import (
	"context"
	"gitee.com/crack007/goose"
	"github.com/crack007/go-im/src/common/config"
	"github.com/crack007/go-im/src/common/constant"
	"github.com/crack007/go-im/src/common/dao"
	"github.com/crack007/go-im/src/common/entity"
	"github.com/crack007/go-im/src/common/model"
	"github.com/crack007/go-im/src/common/service"
	"github.com/crack007/go-im/src/module/api/dto"
	"sort"
)

type messageService struct {
	msgProducer *service.KafkaProducer
	msgDtoQueue chan *dto.MessageDto
}

func (s *messageService) Init(ctx context.Context) {
	s.msgProducer = service.NewKafkaProducer()
	s.msgDtoQueue = make(chan *dto.MessageDto, 1024)
	err := s.msgProducer.Init()
	if err != nil {
		goose.GetLogger().Error(err.Error())
		return
	}
}

func (s *messageService) Publish(messageDto *dto.MessageDto, userModel *model.UserModel) interface{} {
	if userModel != nil {
		messageDto.From = userModel.Id
	}
	s.msgDtoQueue <- messageDto
	return nil
}

func (s *messageService) ListMessage(queryDto *dto.MessageQueryDto) []*dto.MessageDto {
	var list []*dto.MessageDto
	fromUid := queryDto.FromUid
	toUid := queryDto.ToUid
	messageEntities := dao.MessageDao.List(fromUid, toUid, queryDto.MsgId, queryDto.Page, queryDto.Size)
	if messageEntities == nil || len(messageEntities) == 0 {
		return list
	}
	var idList []int
	msgMap := make(map[uint32]*entity.MessageEntity)
	for _, messageEntity := range messageEntities {
		idList = append(idList, int(messageEntity.ID))
		msgMap[messageEntity.ID] = messageEntity
	}
	sort.Ints(idList)
	for _, id := range idList {
		messageEntity := msgMap[uint32(id)]
		list = append(list, &dto.MessageDto{
			Id:          messageEntity.ID,
			From:        messageEntity.FromUID,
			To:          messageEntity.ToUID,
			MessageType: constant.GetMessageType(messageEntity.MsgType),
			CreatedAt:   messageEntity.CreatedAt,
			Data:        messageEntity.Message,
		})
	}
	return list
}

func (s *messageService) Consume() {
	if s.msgDtoQueue == nil {
		panic("msgDtoQueue为空")
	}
	for messageDto := range s.msgDtoQueue {
		messageModel := model.NewMessageModel()
		messageModel.Action = messageDto.Action
		messageModel.MsgId = service.SnowflakeService.NextId()
		messageModel.Data = messageDto.Data
		messageModel.FromUser = messageDto.From
		messageModel.ToUser = messageDto.To
		messageModel.MessageType = messageDto.MessageType
		messageModel.FromNode = config.GetNodeConfig().NodeName
		err := s.msgProducer.Publish(config.KafkaConfig.Topic, messageModel.String())
		if err != nil {
			goose.GetLogger().Error(err)
		}
	}
	// 消息都处理后才能关闭连接
	s.msgProducer.Close()
}
func (s *messageService) Close() {
	close(s.msgDtoQueue)
}
func (s *messageService) GracefulExit(ctx context.Context) {
	go func() {
		<-ctx.Done()
		s.Close()
	}()
}
